package com.example.demo.utils;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.example.demo.mapper.KafkaMessageLogMapper;
import com.example.demo.pojo.KafkaMessageLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

import static com.example.demo.utils.ConstantUtils.FAILED;
import static com.example.demo.utils.ConstantUtils.SENT;


@Component
@Slf4j
public class KafkaSenderUtils {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;


    @Resource
    private KafkaMessageLogMapper kafkaMessageLogMapper;

    /**
     * 发送限价订单到 Kafka（异步执行）
     *
     * @param topic        管道名
     * @param orderMessage 订单信息
     */
    @Async
    public void sendOrderMessage(String topic, String orderMessage) {
        try {
            // 同步发送消息并等待结果
            SendResult<String, String> result = kafkaTemplate.send(topic, orderMessage).get();

            // 检查发送结果
            if (result != null && result.getRecordMetadata() != null) {
                log.info("消息发送成功, Topic: " + result.getRecordMetadata().topic() +
                        ", Partition: " + result.getRecordMetadata().partition() +
                        ", Offset: " + result.getRecordMetadata().offset());
                //发送成功记录发送日志
                KafkaMessageLog kafkaMessageLog = new KafkaMessageLog();
                kafkaMessageLog.setTopic(topic);
                kafkaMessageLog.setMessage(orderMessage);
                kafkaMessageLog.setStatus(SENT);
                kafkaMessageLogMapper.insert(kafkaMessageLog);
            } else {
                log.info("消息发送失败，SendResult 为空");
                //发送失败记录日志
                KafkaMessageLog kafkaMessageLog = new KafkaMessageLog();
                kafkaMessageLog.setTopic(topic);
                kafkaMessageLog.setMessage(orderMessage);
                kafkaMessageLog.setStatus(FAILED);
                kafkaMessageLog.setErrorMessage("消息发送失败，SendResult 为空");
                kafkaMessageLogMapper.insert(kafkaMessageLog);
            }
        } catch (Exception e) {
            log.error("消息发送失败: " + e.getMessage(), e);
            //发送失败记录日志
            KafkaMessageLog kafkaMessageLog = new KafkaMessageLog();
            kafkaMessageLog.setTopic(topic);
            kafkaMessageLog.setMessage(orderMessage);
            kafkaMessageLog.setStatus(FAILED);
            kafkaMessageLog.setErrorMessage(e.getMessage());
            kafkaMessageLogMapper.insert(kafkaMessageLog);
        }
    }


    /**
     * 重新发送失败消息
     */
    @Scheduled(fixedDelay = 18000000)// 每5小时执行一次
    public void retryFailedMessages() {
        //查询发送失败的数据集合
        QueryWrapper<KafkaMessageLog> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", FAILED);
        List<KafkaMessageLog> kafkaMessageLogs = kafkaMessageLogMapper.selectList(queryWrapper);
        if (kafkaMessageLogs.size() == 0) {
            log.info("无发送失败信息！");
        }
        for (KafkaMessageLog aMessageLogs : kafkaMessageLogs) {
            try {
                // 再次尝试发送消息
                SendResult<String, String> result = kafkaTemplate.send(aMessageLogs.getTopic(), aMessageLogs.getMessage()).get();
                // 检查发送结果
                if (result != null && result.getRecordMetadata() != null) {
                    log.info("消息重新发送成功, Topic: " + result.getRecordMetadata().topic() +
                            ", Partition: " + result.getRecordMetadata().partition() +
                            ", Offset: " + result.getRecordMetadata().offset());
                }
                //修改发送状态
                UpdateWrapper<KafkaMessageLog> updateWrapper = new UpdateWrapper<>();
                updateWrapper.eq("id", aMessageLogs.getId()).set("status", SENT);
                kafkaMessageLogMapper.update(null, updateWrapper);
                assert result != null;
                log.info("重行发送成功，发送消息为:" + result.getRecordMetadata());
            } catch (Exception e) {
                log.error("消息重新发送失败: " + e.getMessage(), e);
                // 修改发送状态
                UpdateWrapper<KafkaMessageLog> updateWrapper = new UpdateWrapper<>();
                updateWrapper.eq("id", aMessageLogs.getId())
                        .set("status", FAILED); // 更新状态为失
                kafkaMessageLogMapper.update(null, updateWrapper);
                log.info("重新发送失败！");
            }
        }
    }
}
